package com.tn.openrpc.stomp;

import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.kaazing.gateway.client.protocol.stomp.StompClient;
import com.kaazing.gateway.client.protocol.stomp.StompEvent;

import com.tn.openrpc.TimeoutException;
import com.tn.openrpc.util.Factory;
import com.tn.openrpc.util.FactoryException;

/**
 * A template class that provides common behaviour required over STOMP.
 */
public class StompTemplate
{
  private static final Logger LOGGER = LoggerFactory.getLogger(StompTemplate.class);

  private static final String DEFAULT_ACKNOWLEDGEMENT = "auto";
  private static final long DEFAULT_TIMEOUT = 10 * 1000;
  private static final String HEADER_CORRELATION_ID = "correlation-id";
  private static final String HEADER_ID = "id";
  private static final String HEADER_REPLY_TO = "reply-to";
  private static final long RESPONSE_POLL = 50;

  private StompClient stompClient;
  private Factory<String> idFactory;
  private long timeout;

  /**
   * Creates a new <code>StompTemplate</code>.
   *
   * @param stompClient the stomp client.
   * @param idFactory   the factory used to create the various ids needed by STOMP.
   */
  public StompTemplate(StompClient stompClient, Factory<String> idFactory)
  {
    this.stompClient = stompClient;
    this.idFactory = idFactory;
    this.timeout = DEFAULT_TIMEOUT;
  }

  /**
   * Sets the timeout.
   */
  public void setTimeout(long timeout)
  {
    this.timeout = timeout;
  }

  /**
   * Sends a message containing the <code>body</code> and waits for a response.
   *
   * @param requestDestination  the destination to send to.
   * @param responseDestination the destination to receive the response on.
   * @param receiptId           the receipt id.
   * @param body                the body of the message.
   *
   * @return the body of the response.
   *
   * @throws StompException   if an error occurs interacting with the <code>StompClient</code>.
   * @throws TimeoutException if a response is not received within the permitted time.
   */
  public String send(
    String requestDestination,
    String responseDestination,
    String receiptId,
    String body
  )
    throws StompException, TimeoutException
  {
    Map<String, String> subscribeHeaders = createSubscriptionHeaders();
    stompClient.subscribe(responseDestination, DEFAULT_ACKNOWLEDGEMENT, null, subscribeHeaders);
    LOGGER.debug("Subscribed to response destination: {} with client: {}", responseDestination, stompClient);

    try
    {
      String correlationId = idFactory.newInstance();

      ResponseHandler responseHandler = new ResponseHandler(correlationId);
      stompClient.addStompListener(responseHandler);

      try
      {
        sendRemoteChannelRequest(requestDestination, responseDestination, body, correlationId, receiptId);
        StompEvent responseEvent = waitForResponse(correlationId, responseHandler.getStompEvents());

        return responseEvent.getBodyAsText();
      }
      finally
      {
        stompClient.removeStompListener(responseHandler);
      }
    }
    catch (RuntimeException e)
    {
      if (e.getCause() instanceof StompException)
      {
        throw (StompException)e.getCause();
      }
      else
      {
        throw e;
      }
    }
    catch (FactoryException e)
    {
      throw new StompException("An error occurred creating a correlation id.", e);
    }
    finally
    {
      stompClient.unsubscribe(responseDestination, null, subscribeHeaders);
      LOGGER.debug("Unsubscribed from response destination: {} with client: {}", responseDestination, stompClient);
    }
  }

  /**
   * Waits for a response event to occur.
   */
  private StompEvent waitForResponse(String correlationId, Queue<StompEvent> stompEvents) throws TimeoutException
  {
    long start = System.currentTimeMillis();

    while (start + timeout > System.currentTimeMillis())
    {
      StompEvent response = stompEvents.peek();
      if (response != null)
      {
        if (correlationId.equals(response.getHeaders().get(HEADER_CORRELATION_ID)))
        {
          return stompEvents.poll();
        }
      }

      try
      {
        Thread.sleep(RESPONSE_POLL);
      }
      catch (InterruptedException e)
      {
        //Ignore.
      }
    }

    throw new TimeoutException("Timeout waiting for response.");
  }

  /**
   * Sends the <code>remoteChannelRequest</code>.
   */
  private void sendRemoteChannelRequest(
    String requestDestination,
    String responseDestination,
    String body,
    String correlationId,
    String receiptId
  )
    throws StompException
  {
    try
    {
      String transactionId = idFactory.newInstance();

      Map<String, String> headers = createHeaders(correlationId, responseDestination);
      stompClient.begin(transactionId, null, headers);

      try
      {
        stompClient.send(body, requestDestination, transactionId, receiptId, headers);
        LOGGER.debug("Request sent to: {} with client: {}", requestDestination, stompClient);
      }
      finally
      {
        stompClient.commit(transactionId, null, headers);
      }
    }
    catch (FactoryException e)
    {
      throw new StompException("An error occurred creating the transaction id.", e);
    }
  }

  /**
   * Creates the subscription headers.
   */
  private Map<String, String> createSubscriptionHeaders() throws StompException
  {
    try
    {
      Map<String, String> subscribeHeaders = new HashMap<String, String>();
      subscribeHeaders.put(HEADER_ID, idFactory.newInstance());

      return subscribeHeaders;
    }
    catch (FactoryException e)
    {
      throw new StompException("An error occurred creating the subscription id.", e);
    }
  }

  /**
   * Creates the headers using the <code>correlationId</code> and <code>responseDestination</code>.
   */
  private Map<String, String> createHeaders(String correlationId, String responseDestination)
  {
    Map<String, String> headers = new HashMap<String, String>();

    headers.put(HEADER_CORRELATION_ID, correlationId);
    headers.put(HEADER_REPLY_TO, responseDestination);

    return headers;
  }

  /**
   * An implementation of <code>StompListener</code> that adds the events received to a queue if the correlation ids match.
   */
  private class ResponseHandler extends NullStompListener
  {
    private String correlationId;
    private Queue<StompEvent> stompEvents;

    /**
     * Creates a new <code>ResponseHandler</code> initialized with the <code>correlationId</code> used to filter the events.
     */
    public ResponseHandler(String correlationId)
    {
      this.correlationId = correlationId;
      this.stompEvents = new ConcurrentLinkedQueue<StompEvent>();
    }

    /**
     * Returns the stomp events received by this <code>ResponseHandler</code>.
     */
    public Queue<StompEvent> getStompEvents()
    {
      return stompEvents;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void onMessage(StompEvent stompEvent)
    {
      process(stompEvent);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void onError(StompEvent stompEvent)
    {
      process(stompEvent);
    }

    /**
     * Processes the <code>stompEvent</code>.
     */
    private void process(StompEvent stompEvent)
    {
      if (correlationId.equals(stompEvent.getHeaders().get(HEADER_CORRELATION_ID)))
      {
        while (!stompEvents.offer(stompEvent))
        {
          LOGGER.info("Delayed processing event: {} Internal queue is full, will try again.", stompEvent.getHeaders());
        }
      }
    }
  }
}

